package skewreduce.framework;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.hadoop.mapreduce.JobID;

import skewreduce.framework.physical.PhysicalOp;


public class SchedulerEvent {
    public enum Type {
        COMPLETE,
        FAILED,
        KILLED,
        RUNNING_SLOW,
        ERROR,
        NOOP
    }
    protected final PhysicalOp source;
    protected final Type type;
    protected final long timestamp;
    
    public SchedulerEvent(Type type,PhysicalOp source) {
        timestamp = System.currentTimeMillis();
        this.type = type;
        this.source = source;

        assert source != null;
    }
    public long getTimestamp() { return timestamp; }
    public Type getEventType() { return type; }
    public PhysicalOp getSource() { return source; }
    public EaggPlan getPlan() { return source.getPlan(); }
    public boolean isFailure() {
        return type == Type.FAILED || type == Type.ERROR;
    }

    public int hashCode() { return source.hashCode(); }

    public String toString() {
        StringBuffer buf = new StringBuffer(128);
        buf.append("event[");
        buf.append(type.toString());
        buf.append(',');
        buf.append(source.getID());
        buf.append(',');
        JobID jid = source.getJobID();
        buf.append( ( jid == null ) ? "<nil>" : jid.toString() );
        buf.append(']');
        return buf.toString();
    }

    public static final class Completion extends SchedulerEvent {
        public Completion(PhysicalOp source) { super(Type.COMPLETE,source); }
    }
    public static final class Failed extends SchedulerEvent {
        public Failed(PhysicalOp source) { super(Type.FAILED,source); }
    }
    public static final class Killed extends SchedulerEvent {
        public Killed(PhysicalOp source) { super(Type.KILLED,source); }
    }
    public static final class Noop extends SchedulerEvent {
        public Noop(PhysicalOp source) { super(Type.NOOP,source); }
    }


    /**
     * concrete event instance. directly injected to completion queue,
     * thus imeplement future interface.
     */
    public static final class RunningSlow extends SchedulerEvent
    implements Future<SchedulerEvent> {
        public RunningSlow(PhysicalOp source) {
            super(Type.RUNNING_SLOW,source);
        }
        public boolean cancel(boolean x) { return false; }
        public SchedulerEvent get() { return this; }
        public SchedulerEvent get(long to,TimeUnit unit) { return this; }
        public boolean isCancelled() { return false; }
        public boolean isDone() { return true; }
    }

    public static final class Error extends SchedulerEvent {
        final Throwable e;
        public Error(PhysicalOp source,Throwable e) {
            super(Type.ERROR,source);
            this.e = e;
        }
        public Throwable getCause() { return e; }
    }
}
